package job.relevance.features;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import mapred.example.IndexReader;
import mapred.example.model.Anchors;
import mapred.example.model.Body;
import mapred.example.model.Index;
import mapred.example.model.Title;
import mapred.example.model.Index.Field;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

public class RLV_Features_Mapper extends IndexReader {

    private static final Text EMPTY = new Text("");

    private static Map<String, List<String>> map = null;

    @Override
    public void map(Index index) throws IOException {
        String url = index.get(Field.RAW_URL).toLowerCase();
        if (map.containsKey(url)) {
            for (String q : map.get(url)) {
                this.output(new Text(q), EMPTY);
                this.output(new Text(""), EMPTY);
                output(q.split("\t")[0].split(" "), index);
            }
        }
    }

    private void output(String[] query, Index index) throws IOException {
        Title title = new Title(index.get(Field.TITLE));
        Body body = new Body(index.get(Field.BODY));
        Anchors anchors = new Anchors(index.get(Field.ANCHOR));

        this.output(new Text(title.toRLV(query)), EMPTY);
        this.output(new Text(body.toRLV(query)), EMPTY);
        this.output(new Text(anchors.toRLV(query)), EMPTY);

        this.output(new Text(title.getStr()), EMPTY);
        this.output(new Text(body.getStr()), EMPTY);
        this.output(new Text(anchors.getStr()), EMPTY);
    }

    @Override
    public void setUp() {
        map = new HashMap<String, List<String>>();

        try {
            String list = this.getConfig("list");
            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(list))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                line = line.trim();
                if (!line.isEmpty()) {
                    String url = line.split("\t")[1].toLowerCase();
                    if (!map.containsKey("url")) {
                        map.put(url, new ArrayList<String>());
                    }
                    map.get(url).add(line);
                }
            }
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(255);
        }
    }

    @Override
    public void tearDown() throws IOException {
    }

}
